package com.api.pay.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Objects;

@Configuration
@Slf4j
public class RabbitConfiguration {
    public static final String API_PAY_QUEUE_NAME = "apiPayQueue";
    public static final String API_PAY_EXCHANGE_DLX = "apiPayQueueExchangeDlx";
    public static final String API_PAY_QUEUE_DLX = "apiPayQueueDlx";
    /**
     * 2h
     */
    public static final Integer PAY_MESSAGE_DELAY = 1000 * 60 * 60 * 2;
    public static final String API_PAY_EXCHANGE_NAME = "apiPayExchange";
    public static final String API_DLX_ROUTINGKEY = "order-dlx";
    private static final Integer SEND_MAX_RETRY_COUNT = 3;

    @Bean
    public Queue queue() {
        HashMap<String, Object> hashMap = new HashMap<>();
        hashMap.put("x-message-ttl", PAY_MESSAGE_DELAY);
        hashMap.put("x-dead-letter-exchange", API_PAY_EXCHANGE_DLX);
        hashMap.put("x-dead-letter-routing-key", API_DLX_ROUTINGKEY);
        Queue queue = new Queue(API_PAY_QUEUE_NAME, true, false, false, hashMap);
        return queue;
    }

    @Bean
    public Queue queueDLX() {
        Queue queue = new Queue(API_PAY_QUEUE_DLX, true, false, false);
        return queue;
    }

    @Bean
    public Exchange exchange() {
        ExchangeBuilder exchangeBuilder = ExchangeBuilder.fanoutExchange(API_PAY_EXCHANGE_NAME);
        exchangeBuilder.durable(true);
        exchangeBuilder.delayed();
        return exchangeBuilder.build();
    }

    @Bean
    public Exchange exchangeDLX() {
        ExchangeBuilder exchangeBuilder = ExchangeBuilder.directExchange(API_PAY_EXCHANGE_DLX);
        exchangeBuilder.durable(true);
        return exchangeBuilder.build();
    }

    @Bean
    public Binding binding() {
        Binding binding = new Binding(API_PAY_QUEUE_NAME, Binding.DestinationType.QUEUE, API_PAY_EXCHANGE_NAME, "order", null);
        return binding;
    }

    @Bean
    public Binding bindingDLX() {
        Binding binding = new Binding(API_PAY_QUEUE_DLX, Binding.DestinationType.QUEUE, API_PAY_EXCHANGE_DLX, API_DLX_ROUTINGKEY, null);
        return binding;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //发送方确认机制
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息发送成功");
            } else {
                log.error("消息发送失败{}", cause);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                assert correlationData != null;
                Message message = Objects.requireNonNull(correlationData.getReturned()).getMessage();
                sendRetryMessage(rabbitTemplate, message);
            }
        });
        return rabbitTemplate;
    }

    private static void sendRetryMessage(RabbitTemplate rabbitTemplate, Message message) {
        rabbitTemplate.convertAndSend(API_PAY_EXCHANGE_NAME, "order", message);
    }

    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("user");
        connectionFactory.setPassword("123456");
        return connectionFactory;
    }

}
